fix(collection): allow writes and reads to proceed during Optimize - #614

tests/db/collection_test.cc:3039

• The test currently ignores all Query() errors, which can mask real regressions introduced by the new Optimize concurrency (e.g., schema/version/segment visibility issues). To keep the regression signal strong, only tolerate the specific expected transient failure (ideally by checking status code/type or a known message substring) and record any other error via record_error().

      auto result = collection->Query(query);
      if (!result.has_value()) {
        // Tolerated: a doc admitted by the writing segment's streaming
        // vector index may not yet be visible in its forward store,
        // transiently failing the query. Pre-existing Insert/Query race,
        // unrelated to Optimize.
        continue;
      }

tests/db/collection_test.cc:2944

• Using a fixed sleep_for(200ms) to 'ensure Optimize is running' is prone to flakiness across different machines and build types (Optimize could finish before workers start, or not reach the intended phase). A more deterministic approach would be to add an explicit synchronization point (e.g., a test hook/failpoint or a latch signaled when Optimize enters the compact phase) so the worker threads reliably exercise the intended concurrency window.

  // give the optimizer time to take its locks and start compacting
  std::this_thread::sleep_for(std::chrono::milliseconds(200));

src/db/index/segment/segment_manager.cc:171

• segment_id is not used in this loop. If the project builds with warnings-as-errors, this structured binding can trigger an unused-variable warning on some toolchains. Consider iterating as for (auto &pair : segments_map_) and pushing pair.second, or otherwise avoid binding the unused key.

    for (auto &[segment_id, segment] : segments_map_) {
      segments.push_back(segment);
    }

tests/db/collection_test.cc:3031

• This uses a C-style cast to convert raw float bytes into a std::string. Prefer reinterpret_cast<const char*> here to avoid dropping constness and to make the intent explicit; it also helps prevent compiler warnings in stricter builds.

      query.target_.set_vector(
          std::string((char *)vector.value().data(),
                      vector.value().size() * sizeof(float)));

tests/db/collection_test.cc:2944

• A fixed 200ms sleep makes the test timing-dependent and potentially flaky on slow/fast machines (Optimize may not have started yet, or may have already finished). Prefer a deterministic synchronization point (e.g., a hook/latch indicating compaction has entered phase 2), or at least poll with a bounded wait for an observable state that implies Optimize is running before starting progress assertions.

  // give the optimizer time to take its locks and start compacting
  std::this_thread::sleep_for(std::chrono::milliseconds(200));

src/db/collection.cc:866

• If Segment::Open() fails after MoveDirectory() succeeds, Optimize() returns early leaving an orphaned segment directory at its final path that is not referenced by the manifest/version. Consider adding rollback cleanup (e.g., delete/move-back new_segment_path) on failure paths after the move, and similarly consider cleanup if the later version apply/flush fails.

    if (!FileHelper::MoveDirectory(tmp_segment_path, new_segment_path)) {
      return Status::InternalError("move segment directory failed");
    }
    compact_task.output_segment_meta_->set_id(new_segment_id);

    auto new_segment =
        Segment::Open(path_, *schema_, *compact_task.output_segment_meta_,
                      id_map_, delete_store_, version_manager_,
                      SegmentOptions{true, options_.enable_mmap_});
    if (!new_segment.has_value()) {
      return new_segment.error();
    }

src/db/index/segment/segment_manager.cc:31

• std::unique_lock is defined in <mutex>. Relying on indirect includes via <shared_mutex> is not guaranteed to be portable across standard library implementations. Ensure <mutex> is included in the relevant header/translation unit where std::unique_lock / std::shared_lock are used.

  std::unique_lock<std::shared_mutex> lock(mutex_);

src/db/collection.cc:247

• The PR description mentions adding an optimize_mtx_ to serialize concurrent Optimize() calls, but the implementation introduces maintenance_mtx_ that serializes a broader set of operations (Optimize + DDL + Flush + Close/Destroy). Either update the PR description to match the new behavior/lock, or adjust the code to align with the described optimize_mtx_ scope if the broader serialization is unintended.

  // Serializes maintenance operations (Optimize, schema DDL, Flush, Close
  // and Destroy) without holding schema_handle_mtx_, so a maintenance
  // operation waiting for a running Optimize never becomes a pending
  // exclusive acquirer of the schema lock (which would block new readers).
  // Lock order: maintenance -> schema -> write -> SegmentManager; never
  // acquire maintenance_mtx_ after any of the others.
  mutable std::mutex maintenance_mtx_;

tests/db/collection_test.cc:2944

• This fixed sleep makes the concurrency regression test timing-dependent and potentially flaky (Optimize may still be in phase 1 or may already complete on fast/slow machines). Prefer a deterministic sync point (e.g., a barrier/latch exposed via test hook/observer), or at minimum assert Optimize is still running after the delay and GTEST_SKIP() (or adjust workload) if it already finished.

  // let the optimizer reach the compact phase
  std::this_thread::sleep_for(std::chrono::milliseconds(200));

tests/db/collection_test.cc:3068

• The way ops_during_optimize is counted (if (!optimize_done.load()) after each operation) can undercount operations that overlapped with Optimize but finished right as optimize_done flips, causing intermittent failures. Consider counting using a pre/post sample (e.g., read a local was_optimizing = !optimize_done.load() before the op) or use a separate 'entered_compact_phase' signal to measure progress during the intended window.

  // if any had been blocked for the whole Optimize, no ops could have
  // completed while Optimize was still running
  ASSERT_GE(writer_result.ops_during_optimize, 2);
  ASSERT_GE(fetch_result.ops_during_optimize, 2);
  ASSERT_GE(query_result.ops_during_optimize, 2);

tests/db/collection_test.cc:3123

• This directly calls .value() on the Stats() result without checking has_value(), which will crash the test on error instead of producing an assertion failure with context. Add ASSERT_TRUE(collection->Stats().has_value()) (or store the result and assert) before accessing .value().

  auto stats = collection->Stats().value();

src/db/index/segment/segment_manager.cc:58

• Holding the SegmentManager exclusive lock while calling segment->destroy() can block all concurrent segment readers for the duration of potentially slow I/O, and increases the risk of lock coupling issues if destroy() calls back into code that needs SegmentManager. A safer pattern is to remove (or move-out) the Segment::Ptr under lock, release the lock, then perform destroy() without holding mutex_.

  std::unique_lock<std::shared_mutex> lock(mutex_);
  auto iter = segments_map_.find(segment_id);
  if (iter == segments_map_.end()) {
    return Status::NotFound("Segment not found");
  }

  auto s = iter->second->destroy();
  CHECK_RETURN_STATUS(s);

  segments_map_.erase(iter);
  return Status::OK();

src/db/collection.cc:864

• If MoveDirectory(...) succeeds but Segment::Open(...) fails, Optimize returns early leaving an orphaned moved segment directory on disk that is not referenced by the manifest/version. Consider adding best-effort cleanup on failure (e.g., delete the moved directory or move it back / move it to a quarantine tmp path) to prevent disk leaks and confusing leftovers.

    if (!FileHelper::MoveDirectory(tmp_segment_path, new_segment_path)) {
      return Status::InternalError("move segment directory failed");
    }
    compact_task.output_segment_meta_->set_id(new_segment_id);

    auto new_segment =
        Segment::Open(path_, *schema_, *compact_task.output_segment_meta_,
                      id_map_, delete_store_, version_manager_,
                      SegmentOptions{true, options_.enable_mmap_});
    if (!new_segment.has_value()) {
      return new_segment.error();
    }

src/db/index/segment/segment_manager.cc:144

• The async lambda in this loop uses [&] capture, but it indirectly captures the loop-local segment reference. Because that reference is reassigned each iteration, async tasks can end up operating on the wrong segment. Capture the Segment::Ptr by value in the lambda.

  for (size_t i = 0; i < segments.size(); i += concurrency) {
    size_t end = std::min(i + concurrency, segments.size());

src/db/collection.cc:867

• If MoveDirectory() or Segment::Open() fails, only the current new_segment_path is cleaned up. Any segments already moved/opened earlier in this loop will be left orphaned on disk (not referenced by any manifest), which can accumulate garbage in the collection directory. Consider best-effort cleanup of all previously moved segments on these early-return paths.

    if (!FileHelper::MoveDirectory(tmp_segment_path, new_segment_path)) {
      return Status::InternalError("move segment directory failed");
    }
    compact_task.output_segment_meta_->set_id(new_segment_id);

src/db/index/segment/segment_manager.cc:33

• std::unique_lock<std::shared_mutex> is used in this file, but the translation unit doesn't include <mutex> (it currently only includes <algorithm>, <future>, <thread>, etc.). Relying on transitive includes is fragile and can break builds on some standard library implementations; include <mutex> explicitly.

Status SegmentManager::add_segment(Segment::Ptr segment) {
  if (!segment) {
    return Status::InvalidArgument("Segment is null");
  }

  std::unique_lock<std::shared_mutex> lock(mutex_);
  segments_map_[segment->id()] = segment;
  return Status::OK();